package com.sh.data.engine.domain.shims.iotdb.manager;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.EnumUtil;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.domain.shims.iotdb.domain.DeviceDomain;
import com.sh.data.engine.domain.shims.iotdb.domain.TabletDomain;
import com.sh.data.engine.domain.shims.iotdb.domain.TabletDomain.TabletDataSet;
import com.sh.data.engine.domain.shims.iotdb.domain.TimeseriesDataSetDomain;
import com.sh.data.engine.domain.shims.iotdb.domain.TimeseriesDomain;
import com.sh.data.engine.domain.shims.iotdb.option.IoTDBOptions;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.springframework.beans.BeanUtils;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class IoTDBSessionManager {

    private static final String TIMESTAMP_MEASUREMENT = "time";

    private IoTDBOptions ioTDBOptions;

    public IoTDBSessionManager(IoTDBOptions ioTDBOptions) {
        this.ioTDBOptions = ioTDBOptions;
    }

    public Session getIoTDBSession() {
        Objects.requireNonNull(this.ioTDBOptions, "ioTDBOptions is null");
        List<String> nodes = Arrays.asList(ioTDBOptions.getServer().split(","));
        nodes = nodes.stream().filter(StringUtils::isNotBlank).collect(Collectors.toList());
        Session session = null;
        if (1 == nodes.size()) {
            String[] serverPort = nodes.get(0).split(":");
            session =
                new Session.Builder()
                    .host(serverPort[0])
                    .port(Integer.parseInt(serverPort[1]))
                    .username(ioTDBOptions.getUsername())
                    .password(ioTDBOptions.getPassword())
                    // .version(Version.V_0_13)
                    .build();
        } else {
            session = new Session(nodes, ioTDBOptions.getUsername(), ioTDBOptions.getPassword());
        }
        return session;
    }

    // 测试连接
    public void testConnection() throws IoTDBConnectionException {
        Session session = null;
        try {
            session = getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
        } catch (Exception e) {
            log.error("连接iotdbserver异常，错误信息：{}", e);
            throw new BusinessException(e.getMessage());
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    // 获取存储组
    public List<String> showStorageGroup(String pathPattern) throws Exception {
        Session session = null;
        List<String> storageGroups = Lists.newArrayList();
        try {
            session = this.getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
            String sql = String.join(" ", "show storage group", pathPattern);
            SessionDataSet dataSet = session.executeQueryStatement(sql);
            while (dataSet.hasNext()) {
                RowRecord next = dataSet.next();
                storageGroups.add(next.getFields().get(0).getStringValue());
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
        return storageGroups;
    }

    /**
     * 返回测点信息 用于同步连线
     *
     * @param pathPattern
     * @return
     * @throws Exception
     */
    public List<FieldInfoDomain> getFiledList(String pathPattern) throws Exception {
        Session session = null;
        List<FieldInfoDomain> fieldInfos = Lists.newArrayList();
        try {
            session = this.getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
            String sql = String.join(" ", "show timeseries", pathPattern);
            SessionDataSet dataSet = session.executeQueryStatement(sql);
            while (dataSet.hasNext()) {
                RowRecord next = dataSet.next();
                FieldInfoDomain fieldInfoDomain = new FieldInfoDomain();
                fieldInfoDomain.setFieldName(StringUtils.substringAfterLast(next.getFields().get(0).getStringValue(), "."));
                fieldInfoDomain.setFieldType(next.getFields().get(3).getStringValue());
                fieldInfoDomain.setComment(next.getFields().get(1).getStringValue());
                fieldInfos.add(fieldInfoDomain);
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
        if (CollectionUtil.isNotEmpty(fieldInfos)) {
            FieldInfoDomain fieldInfoDomain = new FieldInfoDomain();
            fieldInfoDomain.setFieldName(TIMESTAMP_MEASUREMENT);
            fieldInfoDomain.setFieldType(TSDataType.INT64.toString());
            fieldInfoDomain.setComment("时序时间戳");
            fieldInfos.add(0, fieldInfoDomain);
        }
        return fieldInfos;
    }

    /**
     * 获取时序
     *
     * @param pathPattern
     * @return
     * @throws Exception
     */
    public List<TimeseriesDomain> showTimeseries(String pathPattern) throws Exception {
        Session session = null;
        List<TimeseriesDomain> timeseriesList = Lists.newArrayList();
        try {
            session = this.getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
            String sql = String.join(" ", "show timeseries", pathPattern);
            SessionDataSet dataSet = session.executeQueryStatement(sql);
            while (dataSet.hasNext()) {
                RowRecord next = dataSet.next();
                TimeseriesDomain timeseries = new TimeseriesDomain();
                timeseries.setTimeseries(next.getFields().get(0).getStringValue());
                timeseries.setAlias(next.getFields().get(1).getStringValue());
                timeseries.setStorageGroup(next.getFields().get(2).getStringValue());
                timeseries.setDataType(
                    EnumUtil.fromString(TSDataType.class, next.getFields().get(3).getStringValue()));
                timeseries.setEncoding(
                    EnumUtil.fromString(TSEncoding.class, next.getFields().get(4).getStringValue()));
                timeseries.setCompression(
                    EnumUtil.fromString(CompressionType.class, next.getFields().get(5).getStringValue()));
                timeseries.setTags(next.getFields().get(6).getStringValue());
                timeseries.setAttributes(next.getFields().get(7).getStringValue());
                timeseriesList.add(timeseries);
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
        return timeseriesList;
    }

    // 获取实体
    public List<DeviceDomain> getDeviceList(String pathPattern) throws Exception {
        Session session = null;
        List<DeviceDomain> devices = Lists.newArrayList();
        try {
            session = this.getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
            pathPattern = Optional.ofNullable(pathPattern).orElse("");
            String sql = String.join(" ", "show devices", pathPattern);
            SessionDataSet dataSet = session.executeQueryStatement(sql);
            while (dataSet.hasNext()) {
                RowRecord next = dataSet.next();
                DeviceDomain deviceDomain = new DeviceDomain();
                deviceDomain.setDevice(next.getFields().get(0).getStringValue());
                deviceDomain.setAligned(next.getFields().get(1).getBoolV());
                devices.add(deviceDomain);
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
        return devices;
    }

    // 创建时序
    public void createTimeseries(TimeseriesDomain timeseriesDomain)
        throws IoTDBConnectionException, StatementExecutionException {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            if (!session.checkTimeseriesExists(timeseriesDomain.getTimeseries())) {
                session.createTimeseries(
                    timeseriesDomain.getTimeseries(),
                    timeseriesDomain.getDataType(),
                    timeseriesDomain.getEncoding(),
                    timeseriesDomain.getCompression());
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    // 创建多个时序
    public void createMultiTimeseries(List<TimeseriesDomain> timeseriesDomains)
        throws IoTDBConnectionException, StatementExecutionException {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            List<String> paths = new ArrayList<>();
            List<TSDataType> tsDataTypes = new ArrayList<>();
            List<TSEncoding> tsEncodings = new ArrayList<>();
            List<CompressionType> compressionTypes = new ArrayList<>();
            timeseriesDomains.stream()
                .forEach(
                    (ts) -> {
                        paths.add(ts.getTimeseries());
                        tsDataTypes.add(ts.getDataType());
                        tsEncodings.add(ts.getEncoding());
                        compressionTypes.add(ts.getCompression());
                    });
            for (String path : paths) {
                if (session.checkTimeseriesExists(path)) {
                    throw new BusinessException("时间序列: " + path + " 已存在，请检查后重试");
                }
            }
            session.createMultiTimeseries(
                paths, tsDataTypes, tsEncodings, compressionTypes, null, null, null, null);
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    // 插入时序数据-单条
    public void insertRecordOneByOne(List<TimeseriesDataSetDomain> dataSets)
        throws IoTDBConnectionException, StatementExecutionException {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            for (TimeseriesDataSetDomain dataSet : dataSets) {
                String deviceId = dataSet.getDeviceId();
                long timestamp =
                    Optional.ofNullable(dataSet.getTimestamp()).orElse(System.currentTimeMillis());
                session.insertRecord(
                    deviceId,
                    timestamp,
                    dataSet.getMeasurements(),
                    dataSet.getDataTypes(),
                    dataSet.getValues());
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    // 插入时序数据-单条 字符串格式
    public void insertStrRecordOneByOne(List<TimeseriesDataSetDomain> dataSets)
        throws IoTDBConnectionException, StatementExecutionException {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            for (TimeseriesDataSetDomain dataSet : dataSets) {
                String deviceId = dataSet.getDeviceId();
                long timestamp =
                    Optional.ofNullable(dataSet.getTimestamp()).orElse(System.currentTimeMillis());
                List<String> values = copyProperties(dataSet.getValues(), String.class);
                session.insertRecord(deviceId, timestamp, dataSet.getMeasurements(), values);
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    /*
     * A Tablet example:
     * device1
     * time s1, s2, s3
     * 1, 1, 1, 1
     * 2, 2, 2, 2
     * 3, 3, 3, 3
     */
    // 批量插入 指定批大小
    public void insertTablet(TabletDomain tabletDomain)
        throws IoTDBConnectionException, StatementExecutionException {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            List<MeasurementSchema> schemaList = tabletDomain.getSchemaList();
            Tablet tablet =
                new Tablet(tabletDomain.getDeviceId(), schemaList, tabletDomain.getMaxRowNumber());
            int rowIndex = 0;
            for (TabletDataSet tabletDataSet : tabletDomain.getTabletDataSets()) {
                long timestamp =
                    Optional.ofNullable(tabletDataSet.getTimestamp()).orElse(System.currentTimeMillis());
                tablet.addTimestamp(rowIndex, timestamp);
                List<Object> values = tabletDataSet.getValues();
                for (int i = 0; i < schemaList.size(); i++) {
                    tablet.addValue(schemaList.get(i).getMeasurementId(), rowIndex, values.get(i));
                }
                if (tablet.rowSize == tablet.getMaxRowNumber()) {
                    session.insertTablet(tablet, true);
                    tablet.reset();
                }
            }
            if (tablet.rowSize != 0) {
                session.insertTablet(tablet);
                tablet.reset();
            }
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    public PreviewDataDomain previewData(String deviceId) throws Exception {
        Session session = null;
        try {
            session = this.getIoTDBSession();
            session.open(false, this.ioTDBOptions.getTimeout() * 1000);
            String sql = String.join(" ", "select * from", deviceId, "limit 10");
            SessionDataSet dataSet = session.executeQueryStatement(sql);
            dataSet.setFetchSize(100);
            List<String> columnTypes = dataSet.getColumnTypes();
            List<String> fieldNameList = dataSet.getColumnNames();
            List<List<Object>> dataList = new LinkedList<>();
            while (dataSet.hasNext()) {
                RowRecord next = dataSet.next();

                List<Object> rowData = new LinkedList<>();
                rowData.add(next.getTimestamp());
                for (int i = 1; i <= fieldNameList.size() - 1; i++) {
                    TSDataType dataType = EnumUtil.fromString(TSDataType.class, columnTypes.get(i));
                    Object val = next.getFields().get(i - 1).getObjectValue(dataType);
                    if (dataType == TSDataType.TEXT) {
                        val = val.toString();
                    }
                    rowData.add(val);
                }
                dataList.add(rowData);
            }
            return PreviewDataDomain.builder().fieldNameList(fieldNameList).dataList(dataList).build();
        } finally {
            if (Objects.nonNull(session)) {
                session.close();
            }
        }
    }

    /**
     * List<?>泛型擦除与替换
     *
     * @param sourceObjList 原对象List
     * @param targetClass   目标类型
     * @param <T>           目标泛型
     * @return List<T>
     */
    private static <T> List<T> copyProperties(List<?> sourceObjList, Class<T> targetClass) {
        if (sourceObjList == null || sourceObjList.isEmpty() || targetClass == null) {
            return Collections.emptyList();
        }

        return sourceObjList.stream()
            .map(e -> copyProperties(e, targetClass))
            .filter(Objects::nonNull)
            .collect(Collectors.toList());
    }

    /**
     * 对象属性拷贝
     *
     * @param sourceObj   源对象
     * @param targetClass 目标对象
     * @param <T>         目标对象泛型
     * @return 拷贝后的对象
     */
    private static <T> T copyProperties(Object sourceObj, Class<T> targetClass) {
        if (sourceObj == null || targetClass == null) {
            return null;
        }
        try {
            T targetObj = targetClass.newInstance();
            BeanUtils.copyProperties(sourceObj, targetObj);
            return targetObj;
        } catch (Exception e) {
            log.error(
                "copy bean error, from {} to {}",
                sourceObj.getClass().getSimpleName(),
                targetClass.getSimpleName(),
                e);
        }
        return null;
    }
}
